package com.chukun.flink.stream.state.savepoint;

import com.chukun.flink.stream.state.operator.CheckpointedMapOperator;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.contrib.streaming.state.RocksDBStateBackend;
import org.apache.flink.runtime.state.StateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.io.IOException;

/**
 * @author chukun
 * @version 1.0.0
 * @description 保存点的基本操作
 * @createTime 2022年05月16日 20:55:00
 */
public class SavePointRunner {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置全局作业的并行度为2
        env.setParallelism(2);
        // 开启检查点机制，执行检查点的操作的间隔为 20s
        env.enableCheckpointing(20000);
        // 设置保存状态后端的地址
        String path = "file:///E:/checkoutpoint/SavePointRunner";
        // 采用rocksDB作为状态后端
        env.setStateBackend(new EmbeddedRocksDBStateBackend());
        env.getCheckpointConfig().setCheckpointStorage(path);

        DataStream<Long> dataStream = env.addSource(new CheckpointedMapOperator.CustomSource());

        DataStream<String> mapStream = dataStream.map(new CheckpointedMapOperator(false, false)).uid("map-id");

        mapStream.print("输出");

        env.execute("SavePointRunner");
    }
}
